package com.ibm.datamodel.timeseries.parser;

import java.io.IOException;
import java.text.ParseException;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Hex;
import org.apache.commons.io.output.ByteArrayOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Throwables;
import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ByteUtil;
import com.ibm.util.DataTypes;
import com.ibm.util.DateParser;
import com.ibm.util.ErrorCode;
import com.ibm.util.exception.CPSException;

/**
 * @author Yaoliang Chen
 * @version Nov 20, 2015
 */
public class TLVParser implements IMessageParser {

	private ITimeSeriesSchema schema;
	private BiMap<Integer, Integer> tlvId2NumMap;
	private int tagLength;
	private int lenLength;
	private DateParser dateParser = new DateParser();

	static private Logger logger = LoggerFactory.getLogger(TLVParser.class);

	private int tsKeyTag;
	private byte[] dataTypes;

	public TLVParser(ITimeSeriesSchema schema, ObjectNode metadata) throws CPSException {
		if(schema == null || metadata == null){
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
					"must specify both schema and parser-meta for tlv");
		}
		
		this.schema = schema;
		this.tlvId2NumMap = HashBiMap.create();
		this.tagLength = 1;
		this.lenLength = 2;
		
		try {
			this.tagLength = metadata.get("tag-length").asInt(1);
		} catch (NullPointerException e) {

		}
		try {
			this.lenLength = metadata.get("len-length").asInt(2);
		} catch (NullPointerException e) {

		}
		if (tagLength > 4 || lenLength > 4) {
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "too big length of tag or len in TLV (max is 4)");
		}
		
		if(schema.getKeyIndex() == -1 || schema.getTimestampIndex() == -1){
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Must specify key and timestamp field in schema for TLV");
		}
		this.dataTypes = schema.getDataTypes();
		for (int i = 0; i < dataTypes.length; i++) {
			switch (this.dataTypes[i]) {
			case DataTypes.DT_LONG:
			case DataTypes.DT_STRING:
			case DataTypes.DT_DOUBLE:
			case DataTypes.DT_BOOLEAN:
				break;
			default:
				throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
						"do not support data type (" + this.dataTypes[i] + ") for tlv");
			}
		}
		JsonNode tagMap = metadata.get("tag-map");
		if (tagMap == null || !tagMap.isObject())
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "need tag-map tag for tlv mapping");
		Iterator<Entry<String, JsonNode>> fields = ((ObjectNode) tagMap).fields();
		while (fields.hasNext()) {
			Map.Entry<java.lang.String, com.fasterxml.jackson.databind.JsonNode> entry = (Map.Entry<java.lang.String, com.fasterxml.jackson.databind.JsonNode>) fields
					.next();
			String tlvTagName = entry.getKey();
			int tlvTagId = entry.getValue().asInt(Integer.MIN_VALUE);
			if (tlvTagId == Integer.MIN_VALUE) {
				throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "values (" + entry.getValue().asText()
						+ ") in tag-map must be a long value not equal to MIN_LONG_VALUE");
			}

			int tagNum = schema.getTagNumber(tlvTagName);
			if (tagNum == -1) {
				throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
						"The name (" + tlvTagName + ") in tag-map must be defined in the schema");
			}

			tlvId2NumMap.put(tlvTagId, tagNum);

			if (schema.getKeyIndex() == tagNum) {
				this.tsKeyTag = tlvTagId;
			}

		}

	}

	@Override
	public Comparable getTimeSeriesKey(byte[] data) throws CPSException {
		int i = 0;
		while (i < data.length) {
			int tag = (int) ByteUtil.getLong(data, i, this.tagLength);
			i += this.tagLength;
			if (i >= data.length) {
				logger.warn("wrong tlv message 1");
				break;
			}

			int len = (int) ByteUtil.getLong(data, i, this.lenLength);
			i += this.lenLength;

			if (i >= data.length) {
				logger.warn("wrong tlv message 2");
				break;
			}

			if (tag == this.tsKeyTag) {
				if (i + len > data.length) {
					logger.warn("wrong tlv message 3");
					break;
				}

				if (len > Long.MAX_VALUE)
					throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "Too big ts id: " + Hex.encodeHexString(data));

				return ByteUtil.getLong(data, i, len);
			}
			i += len;
		}

		throw new CPSException(ErrorCode.PARSE_ERROR_CODE,
				"Cannot find the tskey of the message: " + Hex.encodeHexString(data));

	}

	@Override
	public ParsedDataMessage parseDataMessage(RawDataMessage message) throws CPSException {
		int i = 0;
		byte[] data = message.getMessageBytes();
		ParsedDataMessage parsedMsg = schema.newMessage();
		while (i < data.length) {
			int tag = (int) ByteUtil.getLong(data, i, this.tagLength);
			i += this.tagLength;
			if (i >= data.length) {
				logger.warn("wrong tlv message 1");
				break;
			}

			int len = (int) ByteUtil.getLong(data, i, this.lenLength);
			i += this.lenLength;

			if (i >= data.length) {
				logger.warn("wrong tlv message 2");
				break;
			}

			Integer number = this.tlvId2NumMap.get(tag);
			if (number == null) {
				logger.warn("unknown tag id {}", tag);
				i += len;
				continue;
			}

			switch (this.dataTypes[number]) {
			case DataTypes.DT_LONG:
				long lvalue = ByteUtil.getLong(data, i, Math.min(len, 8));
				parsedMsg.setValue(number, lvalue);
				if (number == this.schema.getTimestampIndex()) {
					parsedMsg.setTimestamp(lvalue);
				} else if (number == this.schema.getKeyIndex()) {
					parsedMsg.setTsKey(lvalue);
				}
				break;
			case DataTypes.DT_STRING:
				byte[] dest = new byte[len];
				System.arraycopy(data, i, dest, 0, len);
				String str = new String(dest);
				if (number == this.schema.getTimestampIndex()) {
					try {
						long ts = dateParser.getTimestamp(str);
						parsedMsg.setValue(number, ts);
						parsedMsg.setTimestamp(ts);
					} catch (ParseException e) {
						logger.warn("cannot parse timestamp " + str, e);
					}

				} else {
					parsedMsg.setValue(number, str);
				}
				break;
			case DataTypes.DT_DOUBLE:
				double dvalue = 0;
				if (len == 4) {
					dvalue = ByteUtil.getFloat(data, i);
				} else if (len == 8) {
					dvalue = ByteUtil.getDouble(data, i);
				} else {
					logger.warn("unknown double length {}", len);
				}
				parsedMsg.setValue(number, dvalue);
				break;
			case DataTypes.DT_BOOLEAN:
				long bvalue = ByteUtil.getLong(data, i, Math.min(len, 8));
				parsedMsg.setValue(number, bvalue != 0);
				break;
			default:
				logger.warn("unknown data type {}", this.dataTypes[number]);
			}

			i += len;
		}

		return parsedMsg;
	}

	@Override
	public byte[] restoreFromDataMessage(ParsedDataMessage message) throws CPSException {
		try (ByteArrayOutputStream baos = new ByteArrayOutputStream();) {

			int len = message.getValueCount();
			for (int i = 0; i < len; ++i) {
				if(message.getValue(i) == null)
					continue;
				Integer tlvTag = tlvId2NumMap.inverse().get(i);
				if (tlvTag == null) {
					logger.warn("unknown tag number {}", i);
					continue;
				}

				baos.write(ByteUtil.longBytes(tlvTag, this.tagLength));

				switch (this.dataTypes[i]) {
				case DataTypes.DT_LONG:
					baos.write(ByteUtil.longBytes(8, this.lenLength));
					baos.write(ByteUtil.longBytes((Long) message.getValue(i), 8));
					break;
				case DataTypes.DT_STRING:
					String str = null;
					if (i == this.schema.getTimestampIndex()) {
						long tstamp = (Long) message.getValue(i);
						str = dateParser.getTimestampString(tstamp);
					} else {
						str = (String) message.getValue(i);
					}
					byte[] strData = str.getBytes();
					baos.write(ByteUtil.longBytes(strData.length, this.lenLength));
					baos.write(strData);
					break;
				case DataTypes.DT_DOUBLE:
					byte[] dbytes = new byte[8];
					baos.write(ByteUtil.longBytes(8, this.lenLength));
					ByteUtil.setDoubleBytes(dbytes, 0, (Double) message.getValue(i));
					baos.write(dbytes);
					break;
				case DataTypes.DT_BOOLEAN:
					baos.write(ByteUtil.longBytes(1, this.lenLength));
					baos.write(((Boolean) message.getValue(i)) ? ByteUtil.longBytes(1, 1) : ByteUtil.longBytes(0, 1));

					break;
				default:
					baos.write(ByteUtil.longBytes(0, this.lenLength));
				}

			}
			return baos.toByteArray();
		} catch (IOException e) {
			logger.error("Cannot assemble the tlv message: " + e.getMessage(), e);
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.ERROR_MESSAGE_TYPE, "Cannot assemble the tlv message");
		}

	}

	@Override
	public ITimeSeriesSchema getSchema() {
		return schema;
	}

}
